package baseFx.common.lock.redis;

import baseFx.common.Assert;
import baseFx.common.AutoCloseable;
import baseFx.common.concurrent.Task;
import baseFx.common.utils.ConvertUtils;
import baseFx.common.utils.FxDate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.*;

class ConcurLocker implements AutoCloseable {
    private static Logger logger = LoggerFactory.getLogger(ConcurLocker.class);
    public static final int HeartbeatSeconds = 30;
    private static final String incLkey = "RCLOCK_LK";
    private static final String incRkey = "RCLOCK_RK";
    private static final String prefix = "|_RCLOCK@";
    private static final long maxInc = 999999999;
    private static boolean isClose;
    private JedisPool pool;
    private String uuid = UUID.randomUUID().toString();
    private String hashkey;
    private String fieldkey;
    private Heartbeat heartbeat;

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> isClose = true));
    }

    public ConcurLocker(JedisPool pool) {
        this.pool = pool;
    }

    public boolean lock(String key, int max) {
        Assert.isNull(hashkey);
        if (max < 1) {
            max = 1;
        }
        String fullkey = prefix + key;
        try (Jedis j = pool.getResource()) {
            while (true) {
                List<String> lks = j.hmget(fullkey, incLkey, incRkey);
                long lk = 0;
                long rk = 0;
                if (lks.size() == 2) {
                    lk = ConvertUtils.change(lks.get(0), long.class);
                    rk = ConvertUtils.change(lks.get(1), long.class);
                }
                if (lk - rk >= max) {
                    return false;
                }
                if (lk > maxInc && rk > maxInc) {
                    lk -= maxInc;
                    rk -= maxInc;
                    Map<String, String> pm = new HashMap<>(2);
                    pm.put(incLkey, Long.toString(lk));
                    pm.put(incRkey, Long.toString(rk));
                    j.hmset(fullkey, pm);
                }
                String fk = lk + "_" + rk;
                long rs = j.hsetnx(fullkey, fk, uuid + "$" + System.currentTimeMillis());
                if (rs == 1) {
                    j.hincrBy(fullkey, incLkey, 1);
                    hashkey = fullkey;
                    fieldkey = fk;
                    heartbeat = new Heartbeat();
                    return true;
                } else {
                    Long ms = ConvertUtils.change(j.hget(fullkey, fk), Long.class);
                    if (ms != null && FxDate.SECONDS * HeartbeatSeconds + ms < System.currentTimeMillis()) {
                        j.hdel(fullkey, fk);
                    }
                }
            }
        }
    }

    @Override
    public void close() {
        if (hashkey != null) {
            heartbeat.cancel();
            heartbeat = null;
            try (Jedis j = pool.getResource()) {
                j.hincrBy(hashkey, incRkey, 1);
                j.hdel(hashkey, fieldkey);
            }
        }
        hashkey = null;
        fieldkey = null;
        pool = null;
    }

    private class Heartbeat implements Runnable {
        private final Object sync = new Object();
        private String fk = fieldkey;
        private String hk = hashkey;

        public Heartbeat() {
            Task.run(this::run);
        }

        @Override
        public void run() {
            Random rd = new Random(System.nanoTime());
            long avg = HeartbeatSeconds * FxDate.SECONDS / 5;
            while (hk != null) {
                Task.delay(avg * 3 + rd.nextInt((int) avg));
                synchronized (sync) {
                    if (hk == null) {
                        break;
                    }
                    try (Jedis j = pool.getResource()) {
                        String s = j.hget(hk, fk);
                        if (s.startsWith(uuid + "$")) {
                            j.hset(hk, fk, uuid + "$" + System.currentTimeMillis());
                        } else {
                            hk = null;
                        }
                    }
                }
            }
        }

        public void cancel() {
            synchronized (sync) {
                hk = null;
            }
        }
    }

    public static void heartbeat(JedisPool pool) {
        if (pool.isClosed()) {
            return;
        }
        Task.run(() -> {
            while (isClose == false) {
                Task.delay(FxDate.SECONDS * 10);
                try (Jedis j = pool.getResource()) {
                    for (String hash : j.keys(prefix + "*")) {
                        for (String key : j.hkeys(hash)) {
                            String val = j.hget(hash, key);
                            if (val != null && val.contains("$")) {
                                String sub = val.substring(val.indexOf("$") + 1);
                                Long m = ConvertUtils.change(sub, Long.class);
                                if (m != null && m + HeartbeatSeconds * FxDate.SECONDS > System.currentTimeMillis()) {
                                    continue;
                                }
                                j.hdel(hash, key);
                                j.hincrBy(hash, incRkey, 1);
                            }
                        }
                    }
                }
            }
        }).onCompleted(t -> {
            if (t.getError() != null) {
                logger.error("并发锁错误", t.getError());
                Task.delay(FxDate.SECONDS * 30);
            }
            if (isClose == false) {
                heartbeat(pool);
            }
        });
    }
}
